package com.pxene.dmp.mapper;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.URLUtils;
import com.pxene.dmp.constant.BaseConstant;

public class GetUserInfoMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

	private Log log = LogFactory.getLog(GetUserInfoMapper.class);

	private URL urlParser;
	private Text resultKey = new Text();
	private LongWritable resultVal = new LongWritable();
	// 调优为128
	private Map<String, Long> cache = new HashMap<String, Long>(128);
	private Set<String> keySet = new HashSet<String>();

	private static final class ADID_TYPE {
		public static final int IDFA = 1;
		public static final int IMEI = 2;
		public static final int ANDROIDID = 3;
		public static final int MAC = 4;
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
		URI[] uris = context.getCacheFiles();
		if (uris != null) {
			BufferedReader br = null;
			for (URI uri : uris) {
				String filename = uri.toString().substring(uri.toString().lastIndexOf("/") + 1);
				log.info(BaseConstant.LOG_PREFIX + "cache file is " + filename);
				// 优化为20M
				br = new BufferedReader(new FileReader(filename), 1024 * 20);
				String line = null;
		        while ((line=br.readLine()) != null) {
		        	String[] datas = line.split("\t");
		        	if (datas.length == 2) {
		        		String[] values = datas[1].split("\\|");
		        		if (values.length == 11 && StringUtils.isNumeric(values[8])) {
		        			cache.put(values[0] + "|" + values[1] + "|" + values[2] 
		        					+ "|" + values[3] + "|" + values[4] + "|" + values[5] 
        							+ "|" + values[6] + "|" + values[7] + "|" + values[9]
									+ "|" + values[10], Long.parseLong(values[8]));
		        		}
		        	}
		        }
		        br.close();
			}
		}
	}

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 9) {
			return;
		}
		String imsi = columns[0];
		String mdn = columns[1];
		String meid = columns[2];
		String domainName = columns[3];
		String url = columns[4];
		String device = columns[5];
		String os = columns[6];
		String province = columns[7];
		String dataType = columns[8];

		try {
			urlParser = new URL(url);
		} catch (MalformedURLException e) {
			return;
		}

		String idfa = "";
		String androidid = "";
		String mac = "";
		String imei = "";
		Map<String, String> map = URLUtils.getQueryMap(urlParser.getQuery());
		// idfa
		if (map.containsKey("deviceid")) {
			idfa = map.get("deviceid").trim().toUpperCase();
		}
		if (map.containsKey("ida")) {
			idfa = map.get("ida").trim().toUpperCase();
		}
		if (map.containsKey("idfa")) {
			idfa = map.get("idfa").trim().toUpperCase();
		}
		if (idfa.matches("[0-9A-F\\-]{36}")) {
			String keyStr = idfa + "|" + ADID_TYPE.IDFA + "|" + imsi + "|" + mdn + "|" + meid 
					+ "|" + device + "|" + os + "|" + domainName + "|" + province + "|" + dataType;
			long count = 0l;
			// 如果存在相同来源的idfa
			if (cache.containsKey(keyStr)) {
				count = cache.get(keyStr);
				keySet.add(keyStr);
			}
			resultKey.set(keyStr + "|" + count);
			resultVal.set(1l);
			context.write(resultKey, resultVal);
		}
		// androidid
		if (map.containsKey("deviceid")) {
			androidid = map.get("deviceid").trim().toUpperCase();
		}
		if (map.containsKey("adrid")) {
			androidid = map.get("adrid").trim().toUpperCase();
		}
		if (map.containsKey("anid")) {
			androidid = map.get("anid").trim().toUpperCase();
		}
		if (map.containsKey("android_id")) {
			androidid = map.get("android_id").trim().toUpperCase();
		}
		if (map.containsKey("androidid")) {
			androidid = map.get("androidid").trim().toUpperCase();
		}
		if (androidid.matches("[0-9A-F]{16}")) {
			String keyStr = androidid + "|" + ADID_TYPE.ANDROIDID + "|" + imsi + "|" + mdn + "|" + meid 
					+ "|" + device + "|" + os + "|" + domainName + "|" + province + "|" + dataType;
			long count = 0l;
			// 如果存在相同来源的androidid
			if (cache.containsKey(keyStr)) {
				count = cache.get(keyStr);
				keySet.add(keyStr);
			}
			resultKey.set(keyStr + "|" + count);
			resultVal.set(1l);
			context.write(resultKey, resultVal);
		}
		// imei
		if (map.containsKey("deviceid")) {
			imei = map.get("deviceid").trim().toUpperCase();
		}
		if (map.containsKey("aduserid")) {
			imei = map.get("aduserid").trim().toUpperCase();
		}
		if (map.containsKey("imei")) {
			imei = map.get("imei").trim().toUpperCase();
		}
		if (imei.matches("[0-9A-F]{14}") || imei.matches("[0-9A-F]{15}")) {
			String keyStr = imei + "|" + ADID_TYPE.IMEI + "|" + imsi + "|" + mdn + "|" + meid 
					+ "|" + device + "|" + os + "|" + domainName + "|" + province + "|" + dataType;
			long count = 0l;
			// 如果存在相同来源的imei
			if (cache.containsKey(keyStr)) {
				count = cache.get(keyStr);
				keySet.add(keyStr);
			}
			resultKey.set(keyStr + "|" + count);
			resultVal.set(1l);
			context.write(resultKey, resultVal);
		}
		// mac
		if (map.containsKey("deviceid")) {
			mac = map.get("deviceid").trim().toUpperCase();
		}
		if (map.containsKey("mac")) {
			mac = map.get("mac").trim().toUpperCase();
		}
		if (mac.matches("[0-9A-F]{12}") || mac.matches("[0-9A-F:\\-]{17}")) {
			String keyStr = formatMac(mac) + "|" + ADID_TYPE.MAC + "|" + imsi + "|" + mdn + "|" + meid 
					+ "|" + device + "|" + os + "|" + domainName + "|" + province + "|" + dataType;
			long count = 0l;
			// 如果存在相同来源的mac
			if (cache.containsKey(keyStr)) {
				count = cache.get(keyStr);
				keySet.add(keyStr);
			}
			resultKey.set(keyStr + "|" + count);
			resultVal.set(1l);
			context.write(resultKey, resultVal);
		}

	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		// 这上个小时出现这次未出现的记录写入
		resultVal.set(0l);
		for (Map.Entry<String, Long> record : cache.entrySet()) {
			String keyStr = record.getKey();
			if (!keySet.contains(keyStr)) {
				resultKey.set(keyStr + "|" + record.getValue());
				context.write(resultKey, resultVal);
			}
		}
	}

	private static String formatMac(String mac) {
		if (mac.length() == 12) {
			StringBuilder sb = new StringBuilder(mac);
			return sb.insert(2, "-").insert(5, "-").insert(8, "-").insert(11, "-").insert(14, "-").toString();
		}
		if (mac.contains(":")) {
			return mac.replace(":", "-");
		}
		return mac;
	}
	
}
